package com.imooc.coupon.service.impl;

import com.alibaba.fastjson.JSON;
import com.imooc.coupon.constant.Constant;
import com.imooc.coupon.constant.CouponStatus;
import com.imooc.coupon.dao.CouponDao;
import com.imooc.coupon.entity.Coupon;
import com.imooc.coupon.exception.CouponException;
import com.imooc.coupon.feign.SettlementClient;
import com.imooc.coupon.feign.TemplateClient;
import com.imooc.coupon.service.IKafkaService;
import com.imooc.coupon.service.IRedisService;
import com.imooc.coupon.service.IUserService;
import com.imooc.coupon.vo.*;
import javafx.util.Pair;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.MapUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
import org.apache.commons.collections4.CollectionUtils;
import org.springframework.util.StringUtils;

import java.math.BigDecimal;
import java.util.*;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * @Date 2021/11/12 16:37
 * @Author wangshbv
 * @Description 用户服务相关的接口实现，所有的操作，都保存在redis中，并通过kafka传递到mysql中
 */
@Slf4j
@Service
public class UserServiceImpl implements IUserService {
    @Autowired
    private CouponDao couponDao;
    @Autowired
    private IRedisService redisService;
    @Autowired
    private TemplateClient templateClient;

    @Autowired
    private SettlementClient settlementClient;

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @Override
    public List<Coupon> findCouponByStatus(Long userId, Integer status)
            throws CouponException {
        List<Coupon> curCached = redisService.getCachedCoupons(userId, status);
        List<Coupon> preTarget = null;
        if (!CollectionUtils.isEmpty(curCached)) {
            log.debug("cache is not empty");
            preTarget = curCached;
        } else {
            log.debug("coupon cache is empty", userId, status);
            List<Coupon> dbCoupons = couponDao.findAllByUserIdAndStatus(userId, CouponStatus.of(status));
            if (CollectionUtils.isEmpty(dbCoupons)) {
                return dbCoupons;
            }
            //填充dbCoupons的templateSDK字段
            Map<Integer, CouponTemplateSDK>
                    ids2TempklateSD = templateClient.findIds2TempklateSDK(
                    dbCoupons.stream().map(Coupon::getId).collect(Collectors.toList())).getData();
            dbCoupons.forEach(dc -> {
                dc.setTemplateSDK(ids2TempklateSD.get(dc.getTemplateId()));
            });
            //数据库中存在记录
            preTarget = dbCoupons;
            //将记录写入缓存
            redisService.addCouponToCache(userId, preTarget, status);
            //将无效优惠卷提出
            preTarget.stream().filter(c -> c.getId() != -1).collect(Collectors.toList());
            //如果取的是可用优惠卷，需要对已过期的做延迟处理
            if (CouponStatus.of(status) == CouponStatus.USEABLE) {
                CouponClassify classify = CouponClassify.classify(preTarget);
                if (!CollectionUtils.isEmpty(classify.getExpired())) {
                    redisService.addCouponToCache(userId, classify.getExpired(), status);
                    CouponKaFkaMessage message = new CouponKaFkaMessage(
                            CouponStatus.EXPIRED.getCode(),
                            classify.getExpired().stream().
                                    map(Coupon::getId).collect(Collectors.toList()));
                    kafkaTemplate.send(Constant.TOPIC, JSON.toJSONString(message));
                }
                return classify.getUsable();
            }
        }
        return preTarget;
    }

    @Override
    public List<CouponTemplateSDK> findAvailableTemplate(Long userId) throws CouponException {
        long curTime = new Date().getTime();
        List<CouponTemplateSDK> templateSDKS = templateClient.findAllUsableTemplate().getData();
        //  过滤掉过期的优惠券模板
        templateSDKS = templateSDKS.stream()
                .filter(t -> t.getRule().getExpiration()
                        .getDeadLine() > curTime).collect(Collectors.toList());
        Map<Integer, Pair<Integer, CouponTemplateSDK>> limit2Template = new HashMap<>(templateSDKS.size());
        templateSDKS.forEach(t->{
            limit2Template.put(t.getId(), new Pair<>(t.getRule().getLimiation(), t));
        });
        List<CouponTemplateSDK> result = new ArrayList<>(limit2Template.size());
        List<Coupon> userUsablkeCoupons = findCouponByStatus(
                userId, CouponStatus.of(CouponStatus.USEABLE.getCode()).getCode());
        Map<Integer,List<Coupon>> templateID2Couopns = userUsablkeCoupons.stream()
                .collect(Collectors.groupingBy(Coupon::getId));
        //根据template的rule判断是否可以领取优惠劵
        limit2Template.forEach((k,v)->{
            CouponTemplateSDK value = v.getValue();
            result.add(value);
        });
        return result;
    }

    /**
     * 1.从templateClient中拿到对应的优惠券，并检查是否过期
     * 2.根据limiation判断用户是否可以领取
     * 3.save to db
     * 4.填充couponTemplateSDK
     * 5.save to cache
     * @param request
     * @return
     * @throws CouponException
     */
    @Override
    public Coupon acquireTemplate(AcquireTemplateRequest request) throws CouponException {
        Map<Integer, CouponTemplateSDK> ids2Tempklate =
                templateClient.findIds2TempklateSDK(
                        Collections.singletonList(request.getTemplateSDK().getId())).getData();
        if(ids2Tempklate.size()<= 0){
            throw new CouponException("can not Acquire template from templateClient");
        }
        //用户是否可以领取到这张优惠券
        List<Coupon> couponUsableStatus = findCouponByStatus(request.getUserId(), CouponStatus.USEABLE.getCode());
        Map<Integer, List<Coupon>> templateId2Coupons = couponUsableStatus.stream().
                collect(Collectors.groupingBy(Coupon::getTemplateId));
        if(templateId2Coupons.containsKey(request.getTemplateSDK().getId())
                && templateId2Coupons.get(request.getTemplateSDK().getId()).size()
                >= request.getTemplateSDK().getRule().getLimiation()){
            log.error("execeed template assign limiation:{}", request.getTemplateSDK().getId());
            throw new CouponException("execeed template assign limiation");
        }
        //尝试去获取优惠劵
        String couponCode = redisService.tryToAcquireCouponFromCache(request.getTemplateSDK().getId());
        if(StringUtils.isEmpty(couponCode)){
            log.error("cannot not acquire coupon  code", request.getTemplateSDK().getId());
            throw new CouponException("cannot not acquire coupon  code");
        }
        Coupon newCoupon = new Coupon(
                request.getTemplateSDK().getId(), request.getUserId(),
                couponCode,CouponStatus.USEABLE
        );
        newCoupon = couponDao.save(newCoupon);
        //填充coupon 对象的CouponTemplateSDK，一定要在放入缓存之前去填充
        newCoupon.setTemplateSDK(request.getTemplateSDK());
        //放入缓存中
        redisService.addCouponToCache(request.getUserId(),
                Collections.singletonList(newCoupon), CouponStatus.USEABLE.getCode());
        return newCoupon;
    }


    /**
     * 结算核销优惠券
     * 这里需要注意，规则相关处理需要由settlement系统来做，当前系统仅仅做业务处理过程
     * @param info
     * @return
     * @throws CouponException
     */
    @Override
    public SettlementInfo settlement(SettlementInfo info) throws CouponException {
        //当没有传递优惠券时，直接返回商品总价
        List<SettlementInfo.CouponAndTemplateInfo> ctInfos = info.getCouponAndTemplateInfos();
        if(CollectionUtils.isEmpty(ctInfos)){
            log.info("empty coupons for settle");
            double goodsSum = 0.0;
            for (GoodsInfo gi :  info.getGoodsInfos()) {
                goodsSum += gi.getPrice() + gi.getCount();
            }
            info.setCost(retain2Decimals(goodsSum));
        }
        //校验传递的优惠券是否是用户自己的
        List<Coupon> coupons = findCouponByStatus(info.getUserId(), CouponStatus.USEABLE.getCode());
        Map<Integer, Coupon> id2Coupon = coupons.stream().
                collect(Collectors.toMap(Coupon::getId, Function.identity()));
        if(MapUtils.isEmpty(id2Coupon) ||
                !CollectionUtils.isSubCollection(ctInfos.stream().
                        map(SettlementInfo.CouponAndTemplateInfo::getId).
                        collect(Collectors.toList()), id2Coupon.keySet())){
            log.info("{}",id2Coupon.keySet());
            log.error("User Coupon Has Some Problem , It is not SubCollection of Coupons");
            throw new CouponException("User Coupon Has Some Problem , It is not SubCollection of Coupons");
        }
        log.debug("current settlement couopns is user's :{}",ctInfos.size());
        List<Coupon> settleCoupons = new ArrayList<>(ctInfos.size());

        ctInfos.forEach(ci -> settleCoupons.add(id2Coupon.get(ci.getId())));
        //通过结算服务获取结算信息
        SettlementInfo processedInfo = settlementClient.computeRule(info).getData();
        if(processedInfo.getEmploy() && CollectionUtils.isNotEmpty(processedInfo.getCouponAndTemplateInfos())){
            log.debug("current settlement couopns is user's :{}",ctInfos.size());
            redisService.addCouponToCache(info.getUserId(), settleCoupons, CouponStatus.USED.getCode());
            //更新db
            CouponKaFkaMessage message = new CouponKaFkaMessage(CouponStatus.USED.getCode(),
                    settleCoupons.stream().map(Coupon::getId).collect(Collectors.toList()));
            kafkaTemplate.send(Constant.TOPIC, JSON.toJSONString(message));
        }
        return processedInfo;
    }

    /**
     * 保留两位小数
     * @param value
     * @return
     */
    private double retain2Decimals(double value){
        //BigDecimal.ROUND_HALF_UP 值四舍五入
        return new BigDecimal(value).setScale(2, BigDecimal.ROUND_HALF_UP).doubleValue();
    }

}
